"""
Case Type   : 发布订阅
Case Name   : 开启基础数据同步,gs_dump备份发布端数据
Create At   : 2023/04/06
Owner       : cr13
Description :
    1.在两个集群创建表
    2.在集群A插入数据
    3.创建发布订阅
    4.发布端备份
    5.两个集群新建数据库
    6.订阅端查询数据是否更新
    7.发布端向新建数据库中导入数据
    8.集群B创建订阅端,订阅新建库
    9.订阅端查询数据是否更新
    10.清理环境
Expect      :
    1.成功
    2.成功
    3.成功
    4.成功
    5.成功
    6.更新
    7.成功
    8.成功
    9.更新
    10.成功
History     : 
"""

import os
import unittest
from yat.test import macro
from yat.test import Node
from testcase.utils.Common import Common
from testcase.utils.Logger import Logger
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant

Primary_SH = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 != Primary_SH.get_node_num(), '非1+2环境不执行')
class Pubsubclass(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info("-----------this is setup-----------")
        self.log.info(f"-----{os.path.basename(__file__)} start-----")
        self.pri_userdb_pub = Node(node='PrimaryDbUser')
        self.pri_rootdb_pub = Node(node='PrimaryRoot')
        self.pri_userdb_sub = Node(node='remote1_PrimaryDbUser')
        self.pri_rootdb_sub = Node(node='remote1_PrimaryRoot')
        self.constant = Constant()
        self.commsh_pub = CommonSH('PrimaryDbUser')
        self.commsh_sub = CommonSH('remote1_PrimaryDbUser')
        self.com_pub = Common()
        self.com_sub = Common('remote1_PrimaryDbUser')
        self.tbname1 = "t_pub_sub_0265_01"
        self.tbname2 = "t_pub_sub_0265_02"
        self.subname1 = "s_pub_sub_0265_01"
        self.subname2 = "s_pub_sub_0265_02"
        self.pubname = "p_pub_sub_0265"
        self.schema = "sc_pub_sub_0265"
        self.tablespace = "ta_pub_sub_0265"
        self.database = "d_pub_sub_0265"
        self.dump_path = os.path.join(
            os.path.dirname(macro.DB_INSTANCE_PATH), 'dump.tar')
        self.parent_path_pub = os.path.dirname(macro.DB_INSTANCE_PATH)
        self.parent_path_sub = os.path.dirname(macro.DB_INSTANCE_PATH_REMOTE1)
        self.port = str(int(self.pri_userdb_pub.db_port) + 1)
        self.wal_level = self.com_pub.show_param("wal_level")
        self.user_param_pub = f'-U {self.pri_userdb_pub.db_user} ' \
            f'-W {self.pri_userdb_pub.db_password}'
        self.user_param_sub = f'-U {self.pri_userdb_sub.db_user} ' \
            f'-W {self.pri_userdb_sub.db_password}'
        cmd = f"cp " \
            f"{os.path.join(macro.DB_INSTANCE_PATH, 'pg_hba.conf')} " \
            f"{os.path.join(self.parent_path_pub, 'pg_hba.conf')};"
        self.log.info(cmd)
        result = self.pri_userdb_pub.sh(cmd).result()
        self.log.info(result)
        cmd = f"cp " \
            f"{os.path.join(macro.DB_INSTANCE_PATH_REMOTE1, 'pg_hba.conf')}" \
            f" {os.path.join(self.parent_path_sub, 'pg_hba.conf')};"
        self.log.info(cmd)
        result = self.pri_userdb_sub.sh(cmd).result()
        self.log.info(result)
        self.hostname = self.pri_userdb_sub.sh('hostname').result().strip()
        text = "step:预置条件,修改pg_hba expect:成功"
        self.log.info(text)
        guc_res = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    replication  {self.pri_userdb_sub.db_user} '
            f'{self.pri_userdb_sub.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败' + text)
        guc_res = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    all  {self.pri_userdb_sub.db_user} '
            f'{self.pri_userdb_sub.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败' + text)
        result = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG, 'wal_level=logical')
        self.assertTrue(result, '执行失败' + text)
        result = self.commsh_pub.restart_db_cluster(True)
        flg = self.constant.START_SUCCESS_MSG in result or 'Degrade' in result
        self.assertTrue(flg, '执行失败' + text)
        guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host    replication  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_pub.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败' + text)
        guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host all  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_pub.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败' + text)

    def test_pub_sub(self):
        text = "step1:在两个集群创建表 expect:成功"
        self.log.info(text)
        sql_cmd = f"drop schema if exists {self.schema} cascade;" \
            f"create schema {self.schema};" \
            f"drop table if exists {self.schema}.{self.tbname1};" \
            f"create table {self.schema}.{self.tbname1}(id1 " \
            f"int primary key,id2 int, id3 int);"
        pub_result = self.commsh_pub.execut_db_sql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(pub_result)
        self.assertIn(self.constant.DROP_SCHEMA_SUCCESS_MSG, pub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.CREATE_SCHEMA_SUCCESS_MSG, pub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, pub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, pub_result,
                      '执行失败' + text)
        sub_result = self.commsh_sub.execut_db_sql(
            sql_cmd, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_result)
        self.assertIn(self.constant.DROP_SCHEMA_SUCCESS_MSG, sub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.CREATE_SCHEMA_SUCCESS_MSG, sub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, sub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, sub_result,
                      '执行失败' + text)
        sql_cmd = f"drop tablespace if exists {self.tablespace};" \
            f"create tablespace {self.tablespace} relative location " \
            f"'test_tblspc';" \
            f"drop table if exists {self.tbname2};create table " \
            f"{self.tbname2}(id int primary key CONSTRAint id_nn not null," \
            f"use_filename varchar2(20),filename varchar2(255),text " \
            f"varchar2(2000)) tablespace {self.tablespace} partition by " \
            f"range(id)(" \
            f"  partition P1 values less than(30)," \
            f"  partition P2 values less than(60)," \
            f"  partition P3 values less than(90)," \
            f"  partition P4 values less than(maxvalue));"
        pub_result = self.commsh_pub.execut_db_sql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(pub_result)
        self.assertIn(self.constant.TABLESPCE_DROP_SUCCESS, pub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.TABLESPCE_CREATE_SUCCESS, pub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, pub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, pub_result,
                      '执行失败' + text)
        sub_result = self.commsh_sub.execut_db_sql(
            sql_cmd, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_result)
        self.assertIn(self.constant.TABLESPCE_DROP_SUCCESS, sub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.TABLESPCE_CREATE_SUCCESS, sub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, sub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, sub_result,
                      '执行失败' + text)

        text = "step2:在集群A插入数据 expect:成功"
        self.log.info(text)
        sql_cmd = f"insert into {self.schema}.{self.tbname1} " \
            f"values(generate_series(1,100), 1, 1);" \
            f"insert into {self.tbname2} values(1, 'first', '%一', '')," \
            f"(60, 'first', '%二', ''),(90, 'first', '%三', ''); "
        pub_result = self.commsh_pub.execut_db_sql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(pub_result)
        self.assertEqual(pub_result.count(self.constant.INSERT_SUCCESS_MSG), 2,
                         '执行失败' + text)

        text = "step3:创建发布订阅 expect:成功"
        self.log.info(text)
        sql_cmd = f"drop publication if exists {self.pubname};" \
            f"create publication {self.pubname} for table " \
            f"{self.schema}.{self.tbname1},{self.tbname2};"
        pub_result = self.commsh_pub.execut_db_sql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(pub_result)
        self.assertIn(self.constant.drop_pub_succ_msg, pub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.create_pub_succ_msg, pub_result,
                      '执行失败' + text)
        result = self.commsh_sub.execute_generate(
            macro.COMMON_PASSWD, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertIn(self.constant.create_keycipher_success, result,
                      '执行失败' + text)
        sql_cmd = f"drop subscription if exists {self.subname1};" \
            f"create subscription {self.subname1} connection " \
            f"'host={self.pri_userdb_pub.db_host} " \
            f"port={self.port} user={self.pri_userdb_pub.db_user} " \
            f"dbname={self.pri_userdb_pub.db_name} " \
            f"password={self.pri_userdb_pub.db_password}' publication " \
            f"{self.pubname};"
        sub_result = self.commsh_sub.execut_db_sql(
            sql_cmd, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_result)
        self.assertIn(self.constant.drop_sub_succ_msg, sub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.create_sub_succ_msg, sub_result, '执行失败'
                      + text)

        text = "step4:发布端备份 expect:成功"
        self.log.info(text)
        pub_result = self.commsh_pub.exec_gs_dump(self.dump_path)
        self.assertTrue(pub_result, '执行失败' + text)

        text = "step5:两个集群新建数据库 expect:成功"
        self.log.info(text)
        sql_cmd = f"drop database if exists {self.database};create database "\
            f"{self.database};"
        pub_result = self.commsh_pub.execut_db_sql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(pub_result)
        self.assertIn(self.constant.DROP_DATABASE_SUCCESS, pub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.CREATE_DATABASE_SUCCESS, pub_result,
                      '执行失败' + text)
        sub_result = self.commsh_sub.execut_db_sql(
            sql_cmd, self.user_param_sub, None,
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_result)
        self.assertIn(self.constant.DROP_DATABASE_SUCCESS, sub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.CREATE_DATABASE_SUCCESS, sub_result,
                      '执行失败' + text)

        text = "step6:订阅端查询数据是否更新 expect:更新"
        self.log.info(text)
        sql_select = f"select pg_sleep(8);select count(*) from " \
            f"{self.schema}.{self.tbname1};select * from {self.tbname2};"
        sub_result = self.commsh_sub.execut_db_sql(
            sql_select, self.user_param_sub, None,
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_result)
        self.assertIn('1 | first        | %一', sub_result, '执行失败' + text)
        self.assertIn('60 | first        | %二', sub_result, '执行失败' + text)
        self.assertIn('90 | first        | %三', sub_result, '执行失败' + text)
        self.assertEqual(str(sub_result.splitlines()[-9]).strip(), '100',
                         '执行失败' + text)

        text = "step7:发布端向新建库中导入数据 expect:成功"
        self.log.info(text)
        cmd = f"source {macro.DB_ENV_PATH};gs_restore -p " \
            f"{self.pri_userdb_pub.db_port} -d {self.database} " \
            f"{self.dump_path}"
        self.log.info(cmd)
        pub_result = self.com_pub.get_sh_result(self.pri_userdb_pub, cmd)
        self.assertIn(self.constant.RESTORE_SUCCESS_MSG, pub_result,
                      '执行失败' + text)

        text = "step8:集群B创建订阅端,订阅新建库 expect:成功"
        self.log.info(text)
        sql_cmd = f"drop schema if exists {self.schema} cascade;" \
            f"create schema {self.schema};" \
            f"drop table if exists {self.schema}.{self.tbname1};" \
            f"create table {self.schema}.{self.tbname1}(id1 " \
            f"int primary key,id2 int, id3 int);"
        sub_result = self.commsh_sub.execut_db_sql(
            sql_cmd, self.user_param_sub, self.database,
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_result)
        self.assertIn(self.constant.DROP_SCHEMA_SUCCESS_MSG, sub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.CREATE_SCHEMA_SUCCESS_MSG, sub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, sub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, sub_result,
                      '执行失败' + text)
        sql_cmd = f"drop table if exists {self.tbname2};create table " \
            f"{self.tbname2}(id int primary key CONSTRAint id_nn not null," \
            f"use_filename varchar2(20),filename varchar2(255),text " \
            f"varchar2(2000)) tablespace {self.tablespace} partition by " \
            f"range(id)(" \
            f"  partition P1 values less than(30)," \
            f"  partition P2 values less than(60)," \
            f"  partition P3 values less than(90)," \
            f"  partition P4 values less than(maxvalue));"
        sub_result = self.commsh_sub.execut_db_sql(
            sql_cmd, self.user_param_sub, self.database,
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_result)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, sub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, sub_result,
                      '执行失败' + text)

        result = self.commsh_sub.execute_generate(
            macro.COMMON_PASSWD, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertIn(self.constant.create_keycipher_success, result,
                      '执行失败' + text)
        sql_cmd = f"drop subscription if exists {self.subname2};" \
            f"create subscription {self.subname2} connection " \
            f"'host={self.pri_userdb_pub.db_host} " \
            f"port={self.port} user={self.pri_userdb_pub.db_user} " \
            f"dbname={self.database} " \
            f"password={self.pri_userdb_pub.db_password}' publication " \
            f"{self.pubname};"
        sub_result = self.commsh_sub.execut_db_sql(
            sql_cmd, self.user_param_sub, self.database,
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_result)
        self.assertIn(self.constant.drop_sub_succ_msg, sub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.create_sub_succ_msg, sub_result, '执行失败'
                      + text)

        text = "step9:订阅端查询数据是否更新 expect:更新"
        self.log.info(text)
        sql_select = f"select pg_sleep(8);select count(*) from " \
            f"{self.schema}.{self.tbname1};select * from {self.tbname2};"
        sub_result = self.commsh_sub.execut_db_sql(
            sql_select, self.user_param_sub, self.database,
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_result)
        self.assertIn('1 | first        | %一', sub_result, '执行失败' + text)
        self.assertIn('60 | first        | %二', sub_result, '执行失败' + text)
        self.assertIn('90 | first        | %三', sub_result, '执行失败' + text)
        self.assertEqual(str(sub_result.splitlines()[-9]).strip(), '100',
                         '执行失败' + text)

    def tearDown(self):
        text = "step10:清理环境 expect:成功"
        self.log.info(text)
        sql_cmd = f"drop subscription if exists {self.subname1};"
        drop_sub_result = self.commsh_sub.execut_db_sql(
            sql_cmd, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(drop_sub_result)
        sql_cmd = f"drop subscription if exists {self.subname2};"
        drop_sub_result1 = self.commsh_sub.execut_db_sql(
            sql_cmd, self.user_param_sub, self.database,
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(drop_sub_result1)
        sql_cmd = f"drop publication if exists {self.pubname};"
        drop_pub_result = self.commsh_pub.execut_db_sql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(drop_pub_result)
        sql_cmd = f"drop schema if exists {self.schema} cascade;drop table " \
            f"if exists {self.tbname2};"
        result_sub = self.commsh_sub.execut_db_sql(
            sql_cmd, self.user_param_sub, self.database,
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result_sub)
        result_pub = self.commsh_pub.execut_db_sql(
            sql_cmd, self.user_param_pub, self.database)
        self.log.info(result_pub)

        sql_cmd = f"drop schema if exists {self.schema} cascade;drop table " \
            f"if exists {self.tbname2};drop database {self.database};" \
            f"drop tablespace if exists {self.tablespace};"
        result_sub1 = self.commsh_sub.execut_db_sql(
            sql_cmd, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result_sub1)
        result_pub1 = self.commsh_pub.execut_db_sql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(result_pub1)
        cmd = f"rm -rf {self.dump_path}"
        self.log.info(cmd)
        sh_result1 = self.com_pub.get_sh_result(self.pri_userdb_pub, cmd)
        cmd = f"mv " \
            f"{os.path.join(self.parent_path_pub, 'pg_hba.conf')} " \
            f"{os.path.join(macro.DB_INSTANCE_PATH, 'pg_hba.conf')}"
        self.log.info(cmd)
        sh_result2 = self.pri_userdb_pub.sh(cmd).result()
        self.log.info(sh_result2)
        cmd = f"mv " \
            f"{os.path.join(self.parent_path_sub, 'pg_hba.conf')} " \
            f"{os.path.join(macro.DB_INSTANCE_PATH_REMOTE1, 'pg_hba.conf')}"
        self.log.info(cmd)
        sh_result3 = self.pri_userdb_sub.sh(cmd).result()
        self.log.info(sh_result3)
        path1 = os.path.join(
            macro.DB_INSTANCE_PATH, 'pg_location/test_tblspc')
        path2 = os.path.join(
            macro.DB_INSTANCE_PATH_REMOTE1, 'pg_location/test_tblspc')
        cmd1 = f"rm -rf {path1}"
        cmd2 = f"rm -rf {path2}"
        sh_result4 = self.pri_userdb_pub.sh(cmd1).result()
        self.log.info(sh_result4)
        sh_result5 = self.pri_userdb_sub.sh(cmd2).result()
        self.log.info(sh_result5)
        result1 = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'wal_level={self.wal_level}')
        self.log.info(result1)
        pub_result = self.commsh_pub.restart_db_cluster(True)
        sub_result = self.commsh_sub.restart_db_cluster(
            True, macro.DB_ENV_PATH_REMOTE1)
        self.assertEqual('', sh_result1, '执行失败' + text)
        self.assertEqual('', sh_result2, '执行失败' + text)
        self.assertEqual('', sh_result3, '执行失败' + text)
        self.assertEqual('', sh_result4, '执行失败' + text)
        self.assertEqual('', sh_result5, '执行失败' + text)
        self.assertTrue(pub_result, '执行失败' + text)
        self.assertTrue(sub_result, '执行失败' + text)
        self.assertTrue(result1, '执行失败' + text)
        self.assertIn(self.constant.DROP_DATABASE_SUCCESS, result_pub1,
                      '执行失败' + text)
        self.assertIn(self.constant.DROP_DATABASE_SUCCESS, result_sub1,
                      '执行失败' + text)
        self.assertIn(self.constant.drop_pub_succ_msg, drop_pub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.drop_sub_succ_msg, drop_sub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.drop_sub_succ_msg, drop_sub_result1,
                      '执行失败' + text)
        self.assertIn(self.constant.DROP_SCHEMA_SUCCESS_MSG, result_pub,
                      '执行失败' + text)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, result_pub,
                      '执行失败' + text)
        self.assertIn(self.constant.DROP_SCHEMA_SUCCESS_MSG, result_sub,
                      '执行失败' + text)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, result_sub,
                      '执行失败' + text)
        self.assertIn(self.constant.DROP_SCHEMA_SUCCESS_MSG, result_pub1,
                      '执行失败' + text)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, result_pub1,
                      '执行失败' + text)
        self.assertIn(self.constant.TABLESPCE_DROP_SUCCESS, result_pub1,
                      '执行失败' + text)
        self.assertIn(self.constant.DROP_DATABASE_SUCCESS, result_pub1,
                      '执行失败' + text)
        self.assertIn(self.constant.DROP_SCHEMA_SUCCESS_MSG, result_sub1,
                      '执行失败' + text)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, result_sub1,
                      '执行失败' + text)
        self.assertIn(self.constant.TABLESPCE_DROP_SUCCESS, result_sub1,
                      '执行失败' + text)
        self.assertIn(self.constant.DROP_DATABASE_SUCCESS, result_sub1,
                      '执行失败' + text)
        self.log.info(f"-----{os.path.basename(__file__)} end-----")

